package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/8/24
 * @Desc: 学大数据，上多易教育
 * <p>
 * 规则id:  rule-3
 * 静态画像条件:  age>20
 * 动态画像条件:  规则上线后，用户做过 k行为，且行为属性 p1的值的平均值> 200
 * 触发条件: 发生 s行为,且属性 p3 = vo
 **/
public class Rule_3_Calculator implements RuleCalculator {
    Table userProfileTable;
    JSONObject jsonObject;

    ValueState<Double> sumState;
    ValueState<Integer> cntState;

    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection hbaseConnection = ConnectionFactory.createConnection(conf);
        userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

        jsonObject = new JSONObject();

        sumState = runtimeContext.getState(new ValueStateDescriptor<Double>("sum-state", Double.class));
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cnt-state", Integer.class));


    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        String eventId = eventBean.getEventId();

        // 动态画像统计
        if ("k".equals(eventId)) {
            String p1ValueStr = eventBean.getProperties().get("p1");
            double p1Value = Double.parseDouble(p1ValueStr);

            // 对sum-state做累加
            Double sumStateValue = sumState.value();
            sumState.update((sumStateValue == null ? 0 : sumStateValue) + p1Value);

            // 对cnt-state加1
            Integer cntStateValue = cntState.value();
            cntState.update((cntStateValue == null ? 0 : cntStateValue) + 1);
        }

        // 触发判断
        if("s".equals(eventId)){

            // 判断动态画像条件是否满足
            if(sumState.value() != null && cntState.value() != null && sumState.value()/ cntState.value() > 200){

                // 判断静态画像条件是否满足
                Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                getParam.addColumn("f".getBytes(),"age".getBytes());

                Result result = userProfileTable.get(getParam);
                byte[] ageBytes = result.getValue("f".getBytes(), "age".getBytes());
                int age = Integer.parseInt(Bytes.toString(ageBytes));

                // 如果满足，则输出命中信息
                if(age > 20){
                    jsonObject.put("user_id",eventBean.getUid());
                    jsonObject.put("timestamp",eventBean.getTimestamp());
                    jsonObject.put("rule_id","rule-3");

                    collector.collect(jsonObject.toJSONString());
                }


            }


        }




    }
}
